package com.meiya.flink.mock;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.Map;
import java.util.HashMap;
import java.util.UUID;

/**
 * 模拟交易数据生成器
 * 根据真实数据格式生成模拟数据并推送到Kafka
 */
public class TransactionDataGenerator {
    private static final Logger logger = LoggerFactory.getLogger(TransactionDataGenerator.class);
    private static final String BOOTSTRAP_SERVERS = "192.168.182.128:9092";
    private static final String TOPIC = "transaction-data";
    private static final int NUM_RECORDS = 1000;
    // 每条记录生成间隔时间 ms
    private static final int DELAY_MS = 1;

    private static final Random random = new Random();
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // 模拟数据的基础信息
    private static final String[] BANK_TYPES = { "中国建设银行", "中国工商银行", "中国农业银行", "中国银行" };
    private static final String[] TRANSACTION_TYPES = { "转账", "消费", "还款" };
    private static final String[] COUNTER_PARTIES = {
            "财付通-微信转账",
            "支付宝（中国）网络技术有限公司",
            "财付通-微信面对面收款",
            "财付通-微信红包",
            "支付宝-还款"
    };

    private static final Map<String, AccountDataGenerator.AccountInfo> accountInfoMap = AccountDataGenerator.getAccountInfoMap();

    // 当前批次的任务记录ID，默认使用AccountDataGenerator中的批次ID
    private static String batchTaskRecordId = AccountDataGenerator.getBatchTaskRecordId();

    /**
     * 向Kafka发送指定数量的交易测试数据
     * @param bootstrapServers Kafka服务器地址
     * @param topic 主题名称
     * @param numRecords 要生成的交易记录数量
     * @param accountMap 账户信息映射，如果为null则使用内部存储的账户信息
     */
    public static void sendTransactionTestData(String bootstrapServers, String topic, int numRecords, Map<String, AccountDataGenerator.AccountInfo> accountMap) {
        // 使用AccountDataGenerator中的批次任务记录ID和clue_id
        batchTaskRecordId = AccountDataGenerator.getBatchTaskRecordId();
        logger.info("使用批次任务记录ID: {}", batchTaskRecordId);
        
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 如果提供了账户映射，则使用它
        Map<String, AccountDataGenerator.AccountInfo> useAccountMap = accountMap != null ? accountMap : accountInfoMap;
        
        // 如果账户映射为空，则生成一些默认账户
        if (useAccountMap.isEmpty()) {
            useAccountMap = generateDefaultAccounts();
        }

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            logger.info("开始生成并发送交易数据到主题: {}", topic);

            for (int i = 0; i < numRecords; i++) {
                String record = generateTransactionRecord(useAccountMap);
                
                // 从记录中提取ukey作为消息键
                String ukey = extractUkeyFromRecord(record);
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, ukey, record);
                
                final int currentCount = i;
                producer.send(producerRecord, (metadata, exception) -> {
                    if (exception == null) {
                        if (currentCount % 100 == 0) {
                            logger.info("交易数据发送成功: topic={}, partition={}, offset={}", 
                                    metadata.topic(), metadata.partition(), metadata.offset());
                        }
                    } else {
                        logger.error("交易数据发送失败: {}", exception.getMessage(), exception);
                    }
                });

                if (i % 100 == 0) {
                    logger.info("已发送 {} 条交易记录", i);
                }

                if (DELAY_MS > 0) {
                    TimeUnit.MILLISECONDS.sleep(DELAY_MS);
                }
            }

            producer.flush();
            logger.info("交易数据生成完成，共发送 {} 条记录", numRecords);
        } catch (Exception e) {
            logger.error("发送交易数据时出错: {}", e.getMessage(), e);
            throw new RuntimeException("发送交易数据失败", e);
        }
    }
    
    /**
     * 从JSON记录中提取ukey字段
     */
    private static String extractUkeyFromRecord(String record) {
        try {
            return objectMapper.readTree(record).get("ukey").asText();
        } catch (Exception e) {
            return String.valueOf(System.currentTimeMillis());
        }
    }
    
    /**
     * 生成默认账户信息，用于测试
     */
    private static Map<String, AccountDataGenerator.AccountInfo> generateDefaultAccounts() {
        Map<String, AccountDataGenerator.AccountInfo> defaultAccounts = new HashMap<>();
        for (int i = 0; i < 5; i++) {
            AccountDataGenerator.AccountInfo info = new AccountDataGenerator.AccountInfo();
            info.clueId = Long.parseLong("101" + String.format("%06d", random.nextInt(999999)));
            info.bankCard = "621700" + String.format("%010d", 1000000000 + i);
            info.bankName = BANK_TYPES[random.nextInt(BANK_TYPES.length)];
            // 添加用户姓名和身份证号
            info.accountName = "测试用户" + i;
            info.idNumber = generateDefaultIdNumber();
            info.crrq = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            info.taskRecordId = batchTaskRecordId;
            defaultAccounts.put(info.bankCard, info);
        }
        return defaultAccounts;
    }
    
    /**
     * 生成默认身份证号
     */
    private static String generateDefaultIdNumber() {
        // 生成18位身份证号
        StringBuilder id = new StringBuilder();
        id.append("652922"); // 区域码
        id.append(String.format("%04d", 1980 + random.nextInt(20))); // 年
        id.append(String.format("%02d", 1 + random.nextInt(12))); // 月
        id.append(String.format("%02d", 1 + random.nextInt(28))); // 日
        id.append(String.format("%04d", random.nextInt(9999))); // 顺序码
        return id.toString();
    }

    /**
     * 生成交易记录
     * @return 生成的交易记录JSON字符串
     */
    private static String generateTransactionRecord() {
        return generateTransactionRecord(accountInfoMap);
    }

    private static String generateTransactionRecord(Map<String, AccountDataGenerator.AccountInfo> useAccountMap) {
        try {
            ObjectNode record = objectMapper.createObjectNode();

            // 从账户映射中随机选择一个账户
            String[] bankCards = useAccountMap.keySet().toArray(new String[0]);
            String bankCard = bankCards[random.nextInt(bankCards.length)];
            AccountDataGenerator.AccountInfo accountInfo = useAccountMap.get(bankCard);

            // 生成基本信息
            String idNumber = accountInfo.idNumber; // 使用账户中的证件号码
            LocalDateTime transTime = generateRandomTime();
            BigDecimal amount = generateAmount();
            BigDecimal balance = generateBalance();
            String direction = amount.compareTo(BigDecimal.ZERO) > 0 ? "进" : "出";
            
            // 生成对手账号，有10%的概率使用已有的银行卡作为对手账号
            String counterPartyAccount;
            String counterPartyName;
            String counterPartyIdNumber = null;
            
            if (random.nextInt(10) == 0 && bankCards.length > 1) {
                // 从已有账户中选择一个不同的账户作为对手
                String otherBankCard;
                do {
                    otherBankCard = bankCards[random.nextInt(bankCards.length)];
                } while (otherBankCard.equals(bankCard));
                
                counterPartyAccount = otherBankCard;
                AccountDataGenerator.AccountInfo counterPartyInfo = useAccountMap.get(otherBankCard);
                counterPartyName = counterPartyInfo.accountName;
                counterPartyIdNumber = counterPartyInfo.idNumber;
            } else {
                // 生成随机对手账号和名称
                counterPartyAccount = generateCounterPartyAccount();
                counterPartyName = COUNTER_PARTIES[random.nextInt(COUNTER_PARTIES.length)];
            }
            
            String transactionType = TRANSACTION_TYPES[random.nextInt(TRANSACTION_TYPES.length)];
            long ukeyValue = random.nextLong();
            String currentTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            
            // 设置交易记录字段 - 按照需求中的transaction字段规范填充
            record.put("jykh", bankCard); // 交易卡号
            record.put("jyzh", bankCard); // 交易账号
            record.put("jyfhm", accountInfo.accountName); // 交易方户名
            record.put("jyfzjhm", idNumber); // 交易方证件号码
            record.put("jysj", transTime.format(dateFormatter)); // 交易时间
            record.put("jyje", amount.abs().toString()); // 交易金额
            record.put("jyye", balance.toString()); // 交易余额
            record.put("sfbz", direction); // 收付标志（出/进）
            record.put("jydszkh", counterPartyAccount); // 交易对手账号
            record.putNull("xjbz"); // 现金标志
            record.put("dshm", counterPartyName); // 对手户名
            
            // 如果对手是已有账户，则设置对手身份证号
            if (counterPartyIdNumber != null) {
                record.put("dssfzh", counterPartyIdNumber); // 对手身份证号
            } else {
                record.putNull("dssfzh"); // 对手身份证号
            }
            
            record.putNull("dskhyh"); // 对手开户银行
            record.put("zysm", generateTransactionDesc()); // 摘要说明
            record.put("jybz", "CNY"); // 交易币种
            record.put("jywdmc", "深圳:建行深圳市分行资金结算与营运管理部"); // 交易网点名称
            record.putNull("jyfsd"); // 交易发生地
            record.put("jysfcg", "1"); // 交易是否成功
            record.putNull("cph"); // 车牌号
            record.putNull("ipdz"); // IP地址
            record.putNull("macdz"); // MAC地址
            record.putNull("dsjyye"); // 对手交易余额
            record.put("jylsh", generateTransactionId()); // 交易流水号
            record.putNull("rzh"); // 入账号
            record.putNull("pzzl"); // 凭证种类
            record.putNull("pzh"); // 凭证号
            record.put("jygyh", "00000000"); // 交易柜员号
            record.put("bz", counterPartyName); // 备注
            record.putNull("shmc"); // 商户名称
            record.put("jylx", transactionType); // 交易类型
            record.put("cxfkjgyy", "成功"); // 查询反馈结果原因
            record.put("clue_id", AccountDataGenerator.getBatchClueId()); // 使用批次级别的clue_id
            record.put("task_record_id", batchTaskRecordId); // 使用批次任务记录ID
            record.put("a_yhlx", accountInfo.bankName); // 银行类型
            record.put("action_type", "0"); // 操作类型（0 原始数据，1 反向数据）
            record.put("from_jykh", direction.equals("进") ? counterPartyAccount : bankCard); // 来源交易卡号
            record.put("to_jydszkh", direction.equals("进") ? bankCard : counterPartyAccount); // 目标交易对手账号
            record.put("crrq", currentTime); // 插入日期
            record.put("ukey", ukeyValue); // UKey(计算Hash后的唯一值)

            return objectMapper.writeValueAsString(record);
        } catch (Exception e) {
            throw new RuntimeException("生成交易记录时出错", e);
        }
    }

    private static LocalDateTime generateRandomTime() {
        // 生成最近7天内的随机时间
        return LocalDateTime.now().minusDays(random.nextInt(7))
                .minusHours(random.nextInt(24))
                .minusMinutes(random.nextInt(60));
    }

    private static BigDecimal generateAmount() {
        // 生成-10000到10000之间的随机金额
        return new BigDecimal(random.nextDouble() * 20000 - 10000)
                .setScale(2, BigDecimal.ROUND_HALF_UP);
    }

    private static BigDecimal generateBalance() {
        // 生成0到100000之间的随机余额
        return new BigDecimal(random.nextDouble() * 100000)
                .setScale(2, BigDecimal.ROUND_HALF_UP);
    }

    private static String generateCounterPartyAccount() {
        StringBuilder account = new StringBuilder("10");
        for (int i = 0; i < 10; i++) {
            account.append(random.nextInt(10));
        }
        return account.toString();
    }

    private static String generateTransactionDesc() {
        String[] descs = { "消费", "转账", "还款", "充值", "提现" };
        return descs[random.nextInt(descs.length)];
    }

    private static String generateTransactionId() {
        return "101001" + random.nextInt(100) + System.currentTimeMillis();
    }

    private static String generateTaskId() {
        // 返回批次任务记录ID，而不是生成新的UUID
        return batchTaskRecordId;
    }

    /**
     * 主方法，用于独立运行测试
     */
    public static void main(String[] args) {
        // 如果是独立运行，则生成新的批次任务记录ID
        if (AccountDataGenerator.getBatchTaskRecordId() == null) {
            batchTaskRecordId = UUID.randomUUID().toString();
        } else {
            batchTaskRecordId = AccountDataGenerator.getBatchTaskRecordId();
        }
        logger.info("使用批次任务记录ID: {}", batchTaskRecordId);
        
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            logger.info("开始生成并发送交易数据...");

            for (int i = 0; i < NUM_RECORDS; i++) {
                String record = generateTransactionRecord();
                producer.send(new ProducerRecord<>(TOPIC, record));

                if (i % 100 == 0) {
                    logger.info("已发送 {} 条记录", i);
                }

                if (DELAY_MS > 0) {
                    TimeUnit.MILLISECONDS.sleep(DELAY_MS);
                }
            }

            logger.info("数据生成完成，共发送 {} 条记录", NUM_RECORDS);
        } catch (Exception e) {
            logger.error("发送数据时出错: {}", e.getMessage(), e);
        }
    }
}